/**
 * Copyright (C) 2007 The AsyncMail Group <asyncmail@googlegroups.com>
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *         http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.googlecode.asyncmail.smtpclient;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.impl.SimpleLog;
import org.apache.mina.common.ExecutorThreadModel;
import org.apache.mina.common.IoFilterChain;
import org.apache.mina.common.IoFilterChainBuilder;
import org.apache.mina.common.IoFuture;
import org.apache.mina.common.IoFutureListener;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoService;
import org.apache.mina.common.IoServiceConfig;
import org.apache.mina.common.IoServiceListener;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.TrafficMask;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.SocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnectorConfig;
import org.springframework.beans.factory.annotation.Required;

import com.googlecode.asyncmail.common.MailAddress;
import com.googlecode.asyncmail.common.TextLineCodecFactory;
import com.googlecode.asyncmail.dnsservice.DNSService;
import com.googlecode.asyncmail.dnsservice.TemporaryResolutionException;
import com.googlecode.asyncmail.queue.MailQueue;
import com.googlecode.asyncmail.queue.MailQueueItem;
import com.googlecode.asyncmail.smtp.filter.ResponseValidationFilter;
import com.googlecode.asyncmail.smtp.filter.SMTPRequestFilter;

public class SMTPDeliveryScheduler implements Runnable {

    private Log logger;

    private MailQueue queue;

    private int timeOut = 120;

    private SocketConnector connector;

    private SMTPDeliveryIoHandler handler;

    private DNSService dns;

    public static final String CURRENT_RECIPIENT = "CURRENT_RECIPIENT";

    public SMTPDeliveryScheduler() {
        SocketConnectorConfig config = new SocketConnectorConfig();
        config.setThreadModel(ExecutorThreadModel.getInstance("ClientThread"));
        config.setConnectTimeout(timeOut);

        // TODO: Make it configurable
        config.setFilterChainBuilder(new IoFilterChainBuilder() {
            public void buildFilterChain(IoFilterChain arg0) throws Exception {
                arg0.addFirst("textline", new ProtocolCodecFilter(
                        new TextLineCodecFactory()));
                arg0.addLast("smtpfilter", new SMTPRequestFilter());

                ResponseValidationFilter validationFilter = new ResponseValidationFilter();
                validationFilter.setLogger(new SimpleLog("FilterLogger"));

            }
        });

        connector = new SocketConnector();
        connector.addListener(new ServiceListener());
        connector.setDefaultConfig(config);

        handler = new SMTPDeliveryIoHandler();

    }

    @Required
    public void setDnsService(DNSService dns) {
        this.dns = dns;
    }

    @Required
    public void setLogger(Log logger) {
        this.logger = logger;
        handler.setLogger(logger);
    }

    @Required
    public void setMailQueue(MailQueue queue) {
        this.queue = queue;
    }

    public void setTimOut(int timeOut) {
        this.timeOut = timeOut;
    }

    @Required
    public void initialize() {
        Thread t = new Thread(this);
        t.setDaemon(true);
        t.setName("SMTPDeliveryThread");
        t.start();
    }

    /**
     * @see java.lang.Runnable#run()
     */
    public void run() {
        while (true) {
            MailQueueItem item = queue.getNextMailQueueItem();

            List<MailAddress> rList = item.getMailEnvelope().getRecipients();

            for (int i = 0; i < rList.size(); i++) {
                MailAddress mailAddress = rList.get(i);
                logger.debug("Try to deliver the email for recipient: "
                        + mailAddress);
                try {
                    Iterator<String> it = dns.getSMTPHostAddresses(mailAddress
                            .getDomainPart());

                    // TODO: Try the next MX on failure
                    if (it.hasNext()) {
                        String address = (String) it.next();
                        IoFuture future = connector.connect(
                                new InetSocketAddress(address, 25), handler);

                        future.addListener(new ConnectFutureListener(queue,
                                item, mailAddress));
                    } else {
                        // TODO: Sent a bounce
                        logger.debug("No mx record found for domain: "
                                + mailAddress.getDomainPart());
                    }

                } catch (TemporaryResolutionException e) {
                    e.printStackTrace();

                    if (item.getNextRetry() == 0) {
                        item.setNextRetry(30000);
                        item.increaseRetries();
                        queue.addMailQueueItem(item);
                    }
                }
            }
        }
    }

    private final class ConnectFutureListener implements IoFutureListener {
        private MailQueueItem qItem;

        private MailAddress recipient;

        private MailQueue queue;

        public ConnectFutureListener(MailQueue queue, MailQueueItem qItem,
                MailAddress recipient) {
            this.qItem = qItem;
            this.recipient = recipient;
            this.queue = queue;
        }

        /**
         * @see org.apache.mina.common.IoFutureListener#operationComplete(org.apache.mina.common.IoFuture)
         */
        public void operationComplete(IoFuture arg0) {
            try {

                IoSession session = arg0.getSession();
                session.setAttribute(MailQueueItem.MAIL_QUEUE_ITEM, qItem);
                session.setAttribute(CURRENT_RECIPIENT, recipient);

                // if I reenable the traffic mask before, it happens that the
                // mail is
                // not yet present when I receive commands.
                arg0.getSession().setTrafficMask(TrafficMask.ALL);

            } catch (RuntimeException e) {
                // Unable to connect.
                e.printStackTrace();

                if (qItem.getNextRetry() == 0) {
                    qItem.setNextRetry(30000);
                    qItem.increaseRetries();
                    queue.addMailQueueItem(qItem);
                }
            }
        }

    }

    private final class ServiceListener implements IoServiceListener {

        public void serviceActivated(IoService arg0, SocketAddress arg1,
                IoHandler arg2, IoServiceConfig arg3) {
            // Nothing todo here
        }

        public void serviceDeactivated(IoService arg0, SocketAddress arg1,
                IoHandler arg2, IoServiceConfig arg3) {
            // Nothing todo here
        }

        public void sessionCreated(IoSession arg0) {
            // Nothing todo here
        }

        /**
         * @see org.apache.mina.common.IoServiceListener#sessionDestroyed(org.apache.mina.common.IoSession)
         */
        @SuppressWarnings("unchecked")
        public void sessionDestroyed(IoSession arg0) {
            MailQueueItem qItem = (MailQueueItem) arg0
                    .getAttribute(MailQueueItem.MAIL_QUEUE_ITEM);
            List<MailAddress> rList = qItem.getMailEnvelope().getRecipients();

            if (rList.isEmpty() == false && qItem.getNextRetry() == 0) {

                String error = (String) arg0
                        .getAttribute(SMTPDeliveryIoHandler.ERROR);
                if (error != null) {
                    if (error.equals("TEMP")) {
                        qItem.setNextRetry(30000);
                        qItem.increaseRetries();

                        // TODO: Better handling of this
                        List<String> list = ((List<String>) arg0
                                .getAttribute(SMTPDeliveryIoHandler.ERROR_MESSAGE));
                        qItem.setErrorMessage(list.get(0));

                        queue.addMailQueueItem(qItem);
                    } else if (error.equals("PERM")) {

                        // TODO: Move to errorprocesser
                        // just remove the recipient by now
                        rList.remove(((MailAddress) arg0
                                .getAttribute("CURRENT_RECIPIENT")));
                    }

                }
            }
        }

    }

}
